Codecov Report

❌ Patch coverage is 62.56281% with 149 lines in your changes missing coverage. Please review.
✅ Project coverage is 64.80%. Comparing base (dc4e957) to head (ac9542e).
⚠️ Report is 6 commits behind head on master.

@@             Coverage Diff              @@
##             master   #18848      +/-   ##
============================================
- Coverage     64.81%   64.80%   -0.01%     
  Complexity     1322     1322              
============================================
  Files          3393     3396       +3     
  Lines        211246   211704     +458     
  Branches      33208    33300      +92     
============================================
+ Hits         136917   137195     +278     
- Misses        63284    63417     +133     
- Partials      11045    11092      +47     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

Really nice piece of work — the additive design is clean, the no-false-negative reasoning is carefully laid out, and the correctness coverage (planner/leaf cap sharing a constant, null-key exclusion on both sides, NaN range omission, graceful mixed-version proto fallback, visitor completeness enforced by the non-default interface method) is thorough. A few things I'd like to discuss before merge, all around sizing/footprint and validation rather than correctness:

1. The exact-IN tier has no footprint cap. maxBytes (16 MB) only guards the bloom path, and maxInSize only chooses bloom-vs-IN for single-key AUTO. For runtime_filter='in', and for any multi-key join (including AUTO multi-key above threshold), we fall to attachDynamicFilter and emit an exact IN of up to maxBuildRows (≈1M) literals per key, AND'd across keys — a multi-MB filter expression embedded in the leaf PinotQuery with no abandon guard. Could we have the exact-IN path honor the same maxBytes abandon as bloom, so there's one consistent footprint ceiling regardless of tier? That keeps the "no new config in v1" stance while closing the gap.

2. Bloom sizing knobs aren't tunable. MAX_IN_SIZE, MAX_BUILD_ROWS, FPP, and MAX_BYTES are fixed constants with no CONFIG_OF_* keys — the only runtime-reachable attachRuntimeFilter hardcodes the defaults (the parameterized overload is test-only). I understand the intent (they affect selectivity/size, not correctness) and I'm fine deferring real config to a follow-up. Just flagging it so it's a conscious decision, and tying back to (1): MAX_BYTES is the one knob that actually bounds footprint, and right now it doesn't apply to the tier that can get largest.

3. No benchmark. This is fundamentally a perf optimization (the motivation is all scan/serialization/network savings), but there's no JMH or cluster measurement — all tests assert result parity, not speedup. Since there's no selectivity gate yet and the feature can regress when the build side is large or the probe is cheap, a benchmark on the canonical large-fact ⋈ small-dim shape — showing the win at high selectivity and the cost at low selectivity / large build — would both justify the feature and tell us how real the hazard in (1) is. pinot-perf already has MSE/join scaffolding to build on. Reasonable to defer for an opt-in v1, but worth at least one data point.

4. Minor / please confirm: NaN handling is correct for the bloom (membership kept, range dropped), but the exact-IN tier just emits IN(probeCol, …, NaN, …) via the reused computeInOperands. Since float/double keys are now first-class here, can you confirm the leaf IN predicate and the MSE hash join agree on NaN = NaN (i.e. either both match it or both drop it)? If they disagree, exact-IN could drop a probe NaN row the join would keep. Likely consistent with the existing SEMI path, but worth a sentence given how carefully NaN is handled elsewhere.

None of these block correctness — the join remaining the source of truth makes the feature safe. (1) and (4) are the ones I'd most like resolved.

Thanks for the thorough review, @gortiz — these are all fair. Addressed below; the branch is updated.

1. Exact-IN footprint cap. Fixed. attachRuntimeFilter now estimates the exact-IN serialized footprint (estimateExactInBytes, one IN list per key over the build rows) and abandons the filter when it would exceed the same maxBytes ceiling the bloom tier honors — so there is now one consistent footprint ceiling regardless of tier (exact IN mode, multi-key, or AUTO). The SEMI-shared attachDynamicFilter is deliberately left untouched: a SEMI join replaces the join with the leaf IN, so it must always emit it. Abandoning is correctness-neutral here (the real hash join still runs). Unit tests cover the abandon-above and keep-below behavior for every storable key type (INT/LONG/FLOAT/DOUBLE/STRING/BYTES/BIG_DECIMAL), the strict footprint boundary (== maxBytes keeps, one byte over abandons), and the multi-key case.

2. Sizing knobs not tunable. Kept as fixed constants for v1, as a conscious decision (and thanks for the explicit nudge). They only affect selectivity/footprint, never correctness, so I'd rather not ship CONFIG_OF_*/query-option surface that isn't wired end-to-end (the leaf would need them threaded through RuntimeFilterNode); a follow-up can add real config + thread it. Tying back to (1): maxBytes now bounds every tier, so the one knob that governs footprint applies uniformly.

3. Benchmark. Added BenchmarkRuntimeFilterJoin in pinot-perf: a 1M-row, unique-key (1:1) self-join that projects a realistically wide probe row (intCol, longCol, doubleCol, strCol, str2Col, str3Col) so the probe-side shuffle — the thing the filter reduces — dominates, which is the stated motivation (a wide fact table shuffled needlessly). A WHERE t2.intCol % buildMod = 0 controls build selectivity; I compare filter=off (baseline, no hint) vs filter=auto (hint-enabled; cluster default stays off) on INT and STRING join keys. buildMod=10000 (~0.01% build → AUTO picks exact IN) is the high-selectivity case where the reducer can shrink the probe; buildMod=1 (100% build → AUTO picks bloom) is the no-reduction case the feature must be left off for. countJoinInt (only the key crosses the network) is a reference for the overhead floor. Numbers:

_{ms/op, AverageTime, 5×5 iterations, 1 fork, 2 servers, JDK 21; ± is the 99.9% CI.}

Takeaways:

• Big, real win where the probe is genuinely expensive to ship through the join — the wide-row STRING-key join with a 0.01% build: 15.9× (139 → 8.7 ms). This is the canonical "fact table shuffled needlessly" case: the reducer cuts the probe to the ~100 matching keys before the hash exchange.
• Break-even where the probe is already cheap — the INT-key join at 1M rows: 7.8 → 8.0 ms, within the measurement noise. The baseline is already at the COUNT floor (the INT probe is cheap to hash/shuffle at this scale), so there's little probe-side cost for the reducer to remove and its overhead just cancels out. The benefit grows with probe scale and per-row key cost; it does not regress here.
• Cost when nothing can be reduced — the 100% build (buildMod=1): 1.08–1.10× slower on the wide selects, and 2.79× on the bare COUNT (where broadcasting/applying the 1M-key bloom dominates an otherwise ~75 ms query and the probe can't shrink because every key matches). This is exactly the hazard you flagged in (1), and why the feature is opt-in and off by default.

The plan is additive — EXPLAIN for the hinted INT case (trimmed):

LogicalJoin(condition=[=($2, $6)], joinType=[inner])
  PinotLogicalExchange(distribution=[hash[2]])
    RuntimeFilterRel(probeKeys=[[2]], buildKeys=[[0]], filterType=[AUTO])
      LogicalProject(...) / PinotLogicalTableScan(...)                 -- probe leaf
      PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])
        LogicalSort(fetch=[1048577])                                   -- maxBuildRows + 1 truncation guard
          LogicalProject(id) / LogicalFilter(=cat,'rare') / TableScan  -- build keys
  PinotLogicalExchange(distribution=[hash[0]])                         -- real hash join, unchanged
    LogicalProject(id) / LogicalFilter(=cat,'rare') / TableScan

The reducer sits on the probe leaf, fed by a pipeline-breaker broadcast of the (capped) build keys; the hash join below is untouched, so it remains the source of truth and the reduction — not a plan change — is what drives the win.

4. NaN in exact-IN. Confirmed they agree. Added testNaNKeyMatchesBaseline: a self-join on a DOUBLE column where ids 0–2 are NaN. The MSE hash join matches NaN = NaN (baseline COUNT(*) = 9: 3 NaN probe × 3 NaN build), and the exact-IN, bloom, and AUTO reducers all reproduce exactly that — so the leaf IN predicate and the join agree on NaN, no probe-NaN row is dropped. (Consistent with the existing SEMI path, now pinned by a test.)